package com.shujia.flink.source
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo02SourceFunction {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //使用自定义source
    val myDS: DataStream[Int] = env.addSource(new MySource)
    myDS.print()

    env.execute()


    /**
     * 自定义source,实现SourceFunction接口
     */


  }
  class MySource extends SourceFunction[Int]{


    /**
     * run方法只执行一次
     * run用于读取数据方法，可以将读取到的数据发送到下游
     *
     * @param ctx ：用于将数据发送到下游
     */
    override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {
      while (true){
        //发送数据到下游
        ctx.collect(1)

        Thread.sleep(100)

      }
    }


    override def cancel(): Unit = {

    }
  }

}
